// Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.

// 异步任务
package db

import (
	"context"
	"sync"
	"time"

	"github.com/beego/beego/v2/client/orm"
	"github.com/pkg/errors"

	"scase.io/application-auto-scaling-service/pkg/common"
	"scase.io/application-auto-scaling-service/pkg/utils"
	"scase.io/application-auto-scaling-service/pkg/utils/logger"
)

const (
	tableNameAsyncTask = "async_task"

	TaskTypeScaleOutScalingGroup  = "scale_out"
	TaskTypeScaleInScalingGroup   = "scale_in"
	TaskTypeDeleteVm              = "delete_vm"
	TaskTypeDeleteScalingGroup    = "delete_scaling_group"
	TaskTypeDeleteNamespace       = "delete_pod_namespace"
	TaskTypeScaleOutEcsGroup      = "e_scale_out"
	TaskTypeScaleInEcsGroup       = "e_scale_in"
	TaskTypeDeleteEcsScalingGroup = "e_delete_group"
	TaskTypeScaleOutPodGroup      = "pod_scale_out"
	TaskTypeScaleInPodGroup       = "pod_scale_in"
	TaskTypeDeletePodGroup        = "pod_delete_group"
	TaskTypeDeletePod             = "delete_pod"
)

type AsyncTask struct {
	Id int
	// 任务类型
	TaskType string `orm:"column(task_type);size(64)"`
	// 任务操作的资源对象的Id
	// 比如扩缩容任务的TaskKey为待扩缩的groupID；vm删除任务的TaskKey为待删除的vmId
	TaskKey   string `orm:"column(task_key);size(128)"`
	TaskId    string `orm:"column(task_id);size(128)"`
	AffectNum int32  `orm:"column(affect_num);default(0)"`
	// 任务配置
	TaskConf string `orm:"column(task_conf);type(text)"`
	// 任务执行节点
	WorkNodeId string `orm:"column(work_node_id);size(128)"`
	TimeModel
}

// TableIndex 设置索引, taskId索引
func (t *AsyncTask) TableIndex() [][]string {
	return [][]string{
		{"TaskType", "TaskKey", "IsDeleted"},
		{"TaskId"},
	}
}

// newAsyncTaskForScaleOut 扩容任务db对象构造方法
func newAsyncTaskForScaleOut(groupId, projectId string, targetNum int32) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeScaleOutScalingGroup,
		TaskKey:  groupId,
		TaskConf: utils.ToJson(&ScaleOutTaskConf{
			groupId,
			targetNum,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForEcsScaleOut 扩容任务db对象构造方法
func newAsyncTaskForEcsGroupScaleOut(taskId string, groupId string, instanceNum int32) *AsyncTask {
	task := &AsyncTask{
		TaskId:     taskId,
		TaskType:   TaskTypeScaleOutEcsGroup,
		TaskKey:    groupId,
		AffectNum:  instanceNum,
		WorkNodeId: common.LocalWorkNodeId,
		TaskConf: utils.ToJson(&EcsGroupScaleOutTaskConf{
			ScalingGroupId: groupId,
			InstanceNum:    instanceNum,
		}),
	}
	task.IsDeleted = notDeletedFlag
	return task
}

func newAsyncTaskForPodGroupScaleOut(taskId string, groupId string, instanceNum int32) *AsyncTask {
	task := &AsyncTask{
		TaskId:     taskId,
		TaskType:   TaskTypeScaleOutPodGroup,
		TaskKey:    groupId,
		AffectNum:  instanceNum,
		WorkNodeId: common.LocalWorkNodeId,
		TaskConf: utils.ToJson(&PodGroupScaleOutTaskConf{
			ScalingGroupId: groupId,
			InstanceNum:    instanceNum,
		}),
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForScaleIn 缩容任务db对象构造方法
func newAsyncTaskForEcsGroupScaleIn(taskId, groupId string, scaleInInstanceIds []string) *AsyncTask {
	task := &AsyncTask{
		TaskId:    taskId,
		TaskType:  TaskTypeScaleInEcsGroup,
		TaskKey:   groupId,
		AffectNum: -int32(len(scaleInInstanceIds)),
		TaskConf: utils.ToJson(&ScaleInTaskConf{
			groupId,
			scaleInInstanceIds,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// 缩容任务db对象构造方法
func newAsyncTaskForPodGroupScaleIn(taskId, groupId string, scaleInPods []string) *AsyncTask {
	task := &AsyncTask{
		TaskId:    taskId,
		TaskType:  TaskTypeScaleInPodGroup,
		TaskKey:   groupId,
		AffectNum: -int32(len(scaleInPods)),
		TaskConf: utils.ToJson(&ScaleInTaskConf{
			groupId,
			scaleInPods,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForScaleIn 缩容任务db对象构造方法
func newAsyncTaskForScaleIn(groupId, projectId string, scaleInInstanceIds []string) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeScaleInScalingGroup,
		TaskKey:  groupId,
		TaskConf: utils.ToJson(&ScaleInTaskConf{
			groupId,
			scaleInInstanceIds,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForDeleteVm vm删除任务db对象构造方法
func newAsyncTaskForDeleteVm(vmId string, groupId string, projectId string) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeDeleteVm,
		TaskKey:  vmId,
		TaskConf: utils.ToJson(&DeleteVmTaskConf{
			vmId,
			groupId,
			projectId,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForDeleteVm vm删除任务db对象构造方法
func newAsyncTaskForDeletePod(podName string, groupId string, projectId string) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeDeletePod,
		TaskKey:  podName,
		TaskConf: utils.ToJson(&DeletePodTaskConf{
			podName,
			groupId,
			projectId,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForDeleteVm 伸缩组删除任务db对象构造方法
func newAsyncTaskForDeleteScalingGroup(groupId string) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeDeleteScalingGroup,
		TaskKey:  groupId,
		TaskConf: utils.ToJson(&DeleteScalingGroupTaskConf{
			groupId,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForDeleteVm 伸缩组删除任务db对象构造方法
func newAsyncTaskForDeleteEcsScalingGroup(taskId, groupId string) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeDeleteEcsScalingGroup,
		TaskKey:  groupId,
		TaskId:   taskId,
		TaskConf: utils.ToJson(&DeleteScalingGroupTaskConf{
			groupId,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// newAsyncTaskForDeletePodNamespace 命名空间删除任务db对象构造方法
func newAsyncTaskForDeletePodNamespace(taskId, groupId string) *AsyncTask {
	task := &AsyncTask{
		TaskType: TaskTypeDeleteNamespace,
		TaskKey:  groupId,
		TaskId:   taskId,
		TaskConf: utils.ToJson(&DeletePodNamespaceTaskConf{
			groupId,
		}),
		WorkNodeId: common.LocalWorkNodeId,
	}
	task.IsDeleted = notDeletedFlag
	return task
}

// InsertDeleteScalingGroupTask ...
// 若该任务已存在（删除伸缩组api的可重入，会导致重复插入的情况），返回错误 ErrDeleteTaskAlreadyExists
func InsertDeleteScalingGroupTask(groupId string) error {
	return ormer.DoTx(func(ctx context.Context, txOrm orm.TxOrmer) error {
		// 1. 判断若task已存在，无需insert，返回错误 ErrAsyncTaskAlreadyExists
		queryTask := &AsyncTask{
			TaskType:  TaskTypeDeleteScalingGroup,
			TaskKey:   groupId,
			TimeModel: TimeModel{IsDeleted: notDeletedFlag},
		}
		err := txOrm.ReadForUpdate(queryTask, "TaskType", "TaskKey", "IsDeleted")
		if err != nil {
			// 忽略task不存在的错误
			if !errors.Is(err, orm.ErrNoRows) {
				return errors.Wrap(err, "db read task err")
			}
		}
		if queryTask.TaskConf != "" {
			return errors.Wrapf(common.ErrDeleteTaskAlreadyExists,
				"delete group task for group[%s] already exists", groupId)
		}

		// 2. 执行task插入操作
		task := newAsyncTaskForDeleteScalingGroup(groupId)
		task.IsDeleted = notDeletedFlag
		_, err = txOrm.Insert(task)
		if err != nil {
			return errors.Wrapf(err, "db add async task[%s:%s] err", task.TaskType, task.TaskKey)
		}
		return nil
	})
}

// InsertDeleteVmAsyncTask ...
func InsertDeleteVmAsyncTask(vmId string, groupId string, projectId string) error {
	// 若该任务已存在，不做操作，返回错误 ErrAsyncTaskAlreadyExists
	exist := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskType, TaskTypeDeleteVm).
		Filter(fieldNameTaskKey, vmId).
		Filter(fieldNameIsDeleted, notDeletedFlag).Exist()
	if exist {
		return errors.Wrapf(common.ErrDeleteTaskAlreadyExists,
			"delete vm task for vm[%s] already exists", vmId)
	}

	task := newAsyncTaskForDeleteVm(vmId, groupId, projectId)
	task.IsDeleted = notDeletedFlag
	_, err := ormer.Insert(task)
	if err != nil {
		return errors.Wrapf(err, "db add async task[%s:%s] err", task.TaskType, task.TaskKey)
	}
	return nil
}

// InsertDeleteVmAsyncTask ...
func InsertDeletePodAsyncTask(podName, groupId, projectId string) error {
	// 若该任务已存在，不做操作，返回错误 ErrAsyncTaskAlreadyExists
	exist := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskType, TaskTypeDeletePod).
		Filter(fieldNameTaskKey, podName).
		Filter(fieldNameIsDeleted, notDeletedFlag).Exist()
	if exist {
		return errors.Wrapf(common.ErrDeleteTaskAlreadyExists,
			"delete pod task for vm[%s] already exists", podName)
	}

	task := newAsyncTaskForDeletePod(podName, groupId, projectId)
	task.IsDeleted = notDeletedFlag
	_, err := ormer.Insert(task)
	if err != nil {
		return errors.Wrapf(err, "db add async task[%s:%s] err", task.TaskType, task.TaskKey)
	}
	return nil
}

// DeleteAsyncTask ...
func DeleteAsyncTask(taskType, taskKey string) error {
	_, err := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskType, taskType).
		Filter(fieldNameTaskKey, taskKey).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Update(orm.Params{
			fieldNameIsDeleted: deletedFlag,
			fieldNameDeleteAt:  time.Now().UTC()})
	if err != nil {
		if errors.Is(err, orm.ErrNoRows) {
			return nil
		}
		return errors.Wrapf(err, "db delete async task[%s:%s] err", taskType, taskKey)
	}
	return nil
}

// GetAllAsyncTasks ...
func GetAllAsyncTasks() ([]*AsyncTask, error) {
	var tasks []*AsyncTask
	_, err := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		All(&tasks)
	if err != nil {
		return nil, errors.Wrapf(err, "get all async task from db err")
	}
	return tasks, nil
}

// GetAsyncTasksByWorkNodeId ...
func GetAsyncTasksByWorkNodeId(wnId string) ([]*AsyncTask, error) {
	var tasks []*AsyncTask
	_, err := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Filter(fieldNameWorkNodeId, wnId).
		All(&tasks)
	if err != nil {
		return nil, errors.Wrapf(err, "get all async task of work node[%s] from db err", wnId)
	}
	return tasks, nil
}

func GetAsyncTaskByTaskId(taskId string) (*AsyncTask, error) {
	var task AsyncTask
	err := ormer.QueryTable(tableNameAsyncTask).Filter(fieldNameTaskId, taskId).Filter(fieldNameIsDeleted, notDeletedFlag).One(&task)
	if err != nil {
		return nil, err
	}
	return &task, nil
}

func UpdateAsyncTaskAffectNum(taskId string, affectNum int32) error {
	_, err := ormer.QueryTable(tableNameAsyncTask).Filter(fieldNameTaskId, taskId).
		Update(orm.Params{
			"affect_num": affectNum,
		})
	return err
}

// TakeOverAsyncTasks ...
func TakeOverAsyncTasks(taskIds []int, takeOverId string) error {
	if len(taskIds) == 0 {
		return nil
	}
	_, err := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameIdIn, taskIds).
		Update(orm.Params{
			fieldNameWorkNodeId: takeOverId,
		})
	return err
}

// ScaleOutTaskConf 扩容任务配置
type ScaleOutTaskConf struct {
	ScalingGroupId       string `json:"scaling_group_id"`
	TargetInstanceNumber int32  `json:"target_instance_number"`
}

// ScaleInTaskConf 缩容任务配置
type ScaleInTaskConf struct {
	ScalingGroupId     string   `json:"scaling_group_id"`
	ScaleInInstanceIds []string `json:"scale_in_instance_ids"`
}

type EcsGroupScaleOutTaskConf struct {
	ScalingGroupId string `json:"scaling_group_id"`
	InstanceNum    int32  `json:"instance_num"`
	JobId          string `json:"job_id"`
}

type PodGroupScaleOutTaskConf struct {
	ScalingGroupId string `json:"scaling_group_id"`
	InstanceNum    int32  `json:"instance_num"`
}

// DeleteVmTaskConf vm删除任务配置
type DeleteVmTaskConf struct {
	// 待删除的vmId
	DeleteVmId string `json:"delete_vm_id"`
	GroupId    string `json:"group_id"`
	ProjectId  string `json:"project_id"`
}

// pod删除任务配置
type DeletePodTaskConf struct {
	// 待删除的vmId
	PodName   string `json:"pod_name"`
	GroupId   string `json:"group_id"`
	ProjectId string `json:"project_id"`
}

// DeleteScalingGroupTaskConf 伸缩组删除任务配置
type DeleteScalingGroupTaskConf struct {
	ScalingGroupId string `json:"scaling_group_id"`
}

// DeletePodNamespaceTaskConf 命名空间删除任务配置
type DeletePodNamespaceTaskConf struct {
	ScalingGroupId string `json:"scaling_group_id"`
}

// txInsertAsyncTask ...
func txInsertAsyncTask(txOrm orm.TxOrmer, task *AsyncTask) error {
	if task == nil {
		return errors.New("task provided to txInsertAsyncTask must not be nil")
	}

	task.IsDeleted = notDeletedFlag
	_, err := txOrm.Insert(task)
	if err != nil {
		return errors.Wrapf(err, "db add async task[%s:%s] err", task.TaskType, task.TaskKey)
	}
	return nil
}

// txDeleteAsyncTask ...
func txDeleteAsyncTask(txOrm orm.TxOrmer, taskType, taskKey string) error {
	num, err := txOrm.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskType, taskType).
		Filter(fieldNameTaskKey, taskKey).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Update(orm.Params{
			fieldNameIsDeleted: deletedFlag,
			fieldNameDeleteAt:  time.Now().UTC()})
	if err != nil {
		return errors.Wrapf(err, "db delete async task[%s:%s] err", taskType, taskKey)
	}
	if num != 1 {
		logger.R.Warn("There may be a program logic error, updated num[%d]", num)
	}
	return nil
}

func txDeleteAsyncTaskByTaskId(txOrm orm.TxOrmer, taskId string) error {
	num, err := txOrm.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskId, taskId).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Update(orm.Params{
			fieldNameIsDeleted: deletedFlag,
			fieldNameDeleteAt:  time.Now().UTC()})
	if err != nil {
		return errors.Wrapf(err, "db delete async taskId[%s] err", taskId)
	}
	if num != 1 {
		logger.R.Warn("There may be a program logic error, updated num[%d]", num)
	}
	return nil
}

// 查询是否还有同时进行的其他任务
func checkExistTaskKey(txOrm orm.TxOrmer, taskKey string) bool {
	exist := txOrm.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskKey, taskKey).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Exist()
	return exist
}

// 获取group所有正在运行的任务中扩容的数目
func GetAllTaskAffectNumOfGroup(groupId string) int32 {
	var allTask []AsyncTask
	var affectCount int32
	_, err := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskKey, groupId).
		Filter(fieldNameTaskTypeIn, TaskTypeScaleInEcsGroup, TaskTypeScaleOutEcsGroup, TaskTypeScaleInPodGroup, TaskTypeScaleOutPodGroup).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		All(&allTask)
	if err != nil {
		return 0
	}
	for _, task := range allTask {
		affectCount += task.AffectNum
	}
	if affectCount <= 0 {
		return 0
	}
	return affectCount
}

func GetAsyncTaskCountOfTaskKey(taskKey string) (int, error) {
	count, err := ormer.QueryTable(tableNameAsyncTask).
		Filter(fieldNameTaskKey, taskKey).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Count()
	return int(count), err
}

func GetScaleOutTaskJobId(taskId string) string {
	var task AsyncTask
	if err := ormer.QueryTable(tableNameAsyncTask).Filter(fieldNameTaskId, taskId).One(&task); err != nil {
		return ""
	}
	taskConf := &EcsGroupScaleOutTaskConf{}
	if err := utils.ToObject([]byte(task.TaskConf), taskConf); err != nil {
		return ""
	}
	return taskConf.JobId
}

func AddJobIdToScaleOutTask(jobId, taskId string) error {
	var task AsyncTask
	if err := ormer.QueryTable(tableNameAsyncTask).Filter(fieldNameTaskId, taskId).One(&task); err != nil {
		return err
	}
	taskConf := &EcsGroupScaleOutTaskConf{}
	if err := utils.ToObject([]byte(task.TaskConf), taskConf); err != nil {
		return err
	}
	taskConf.JobId = jobId
	task.TaskConf = utils.ToJson(taskConf)
	_, err := ormer.Update(&task)
	if err != nil {
		return err
	}
	return nil
}

func DeepDeleteAsyncTask(id int) error {
	_, err := ormer.QueryTable(tableNameAsyncTask).Filter(fieldNameId, id).Delete()
	return err
}

func CleanUpAsyncTask(dayBefore int, log *logger.FMLogger) error {
	logger.R.Info("start to clean async task table ...")
	timeLimit := time.Now().UTC().AddDate(0, 0, -dayBefore).Format(common.TimeLayout)
	condition := orm.NewCondition()
	condition = condition.And(fieldNameUpdateAtLt, timeLimit)
	var atList = &[]AsyncTask{}
	// 较多，防止一次删除占满资源，添加limit
	count, err := ormer.QueryTable(tableNameAsyncTask).SetCond(condition).Limit(common.DefaultLimit).All(atList)
	if err != nil {
		log.Error("get async task table before %d days ago data with error:%+v", dayBefore, err)
		return err
	}

	wg := sync.WaitGroup{}
	wg.Add(int(count))
	for index, asyncTask := range *atList {
		if index%common.InstanceCountToSleep == 0 {
			time.Sleep(common.InstanceCountToSleepSeconds)
		}
		go func(asyncId int, wg *sync.WaitGroup) {
			defer wg.Done()
			if err := DeepDeleteAsyncTask(asyncId); err != nil {
				log.Error("delete async task [%s] error:%+v", asyncId, err)
			}
			log.Info("delete async task [%s] success", asyncId)
		}(asyncTask.Id, &wg)
	}
	wg.Wait()
	return nil
}
